package com.aotain.project.gora.test;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.Map.Entry;

import org.apache.commons.codec.binary.Base64;
import org.apache.gora.store.DataStore;
import org.apache.gora.store.DataStoreFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.log4j.Logger;
import org.apache.storm.Config;
import org.apache.storm.Constants;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.FailedException;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;

import com.aotain.hbase.HBaseRecordAdd;
import com.aotain.project.gora.generated.SdsAbnormalLog;
import com.aotain.storm.AbstractBolt;

public class GoraHbaseBolt extends AbstractBolt{


	/**
	 * 上行包字节数
	 */
	private HashMap<String,Integer> _mapStreamoctets = new HashMap<String,Integer>();

	/**
	 * 上行包数
	 */
	private HashMap<String,Integer> _mapStreampacket = new HashMap<String,Integer>();

	private static final long serialVersionUID = 1L;
	private Map<String,String> tupleMap = new HashMap<String,String>();
	private Map<String,Integer> sqlNumMap = new HashMap<String,Integer>();
	private int threshold = 3;
	public String injectionKeyList = null;

	/**
	 * 固定源60s内握手成功次数  目标|源|端口
	 */
	private HashMap<String,Integer> fixedSourceMap = new HashMap<String,Integer>();

	/**
	 * 伪造源60s内握手成功次数  目标|端口
	 */
	private HashMap<String,Integer> forgedSourceMap = new HashMap<String,Integer>();

	/**
	 * 上行包字节数
	 */
	private HashMap<String,Integer> upStreamoctets = new HashMap<String,Integer>();

	/**
	 * 上行包数
	 */
	private HashMap<String,Integer> upStreampacket = new HashMap<String,Integer>();

	/**
	 * 受攻击的记录map
	 */
	private HashMap<String,String> abnormalStreamMap = new HashMap<String,String>();
	
	DataStore<String,SdsAbnormalLog> hStore ;

	public GoraHbaseBolt(){
		this.threshold = 5;
	}

	@Override
	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		declarer.declare(new Fields("dip","sip","dport","sport","dateStr_d","dateStr_h","sourceAreaId","sourceAreaName","areaName","pv","dateStr_hour", "abntype", "idcno", "sourcecountry","streamoctets","streampacket"));
		declarer.declareStream("ABNORMAL_LOG", new Fields("rowkey","dip","sip","dport","accesstime","sourcearea",
				"sourcegeo","sourcecountry","destarea","destgeo","evaluate","abnormal","desc","attnum","delaytime","flag"));
	}

	@Override
	public Map<String, Object> getComponentConfiguration() {		
		Map<String,Object> conf = new HashMap<String,Object>();
		conf.put(Config.TOPOLOGY_TICK_TUPLE_FREQ_SECS,60);
		return conf;
	}

	//计算需要的值
	private void countInLocal(Tuple tuple) {
		try
		{
			String dip = tuple.getStringByField("dip");
			String sip = tuple.getStringByField("sip");
			String idc = tuple.getStringByField("idcno");
			String type = tuple.getStringByField("type");
			Integer status = tuple.getIntegerByField("status");
			String gis =  new String(Base64.decodeBase64(tuple.getStringByField("gisstr")));
			String dport = tuple.getStringByField("dport");
			Integer upStreamOctets = tuple.getIntegerByField("upstreamoctets");
			Integer upStreamPacket = tuple.getIntegerByField("upstreampacket");


			int fixedSourceCount = 1;
			String fixedMapKey = dip + "|" + sip + "|" + dport + "|" + gis + "|" + idc;
			//如果是tcp 且握手成功 则计数
//			if(status == 1 && "6".equals(type)) {

				if(fixedSourceMap.containsKey(fixedMapKey)) {
					fixedSourceCount = fixedSourceMap.get(fixedMapKey) + 1;
				}

				if(upStreamoctets.containsKey(fixedMapKey)) {
					upStreamOctets += upStreamoctets.get(fixedMapKey) ;
				}

				if(upStreampacket.containsKey(fixedMapKey)) {
					upStreamPacket += upStreampacket.get(fixedMapKey);
				}
				fixedSourceMap.put(fixedMapKey, fixedSourceCount);
				upStreamoctets.put(fixedMapKey, upStreamOctets);
				upStreampacket.put(fixedMapKey, upStreamPacket);

//			}
		}
		catch(Exception ex)
		{
			Logger.getRootLogger().error("=====countInLocal=========ERROR=================",ex);
		}
	}



	//定时发送
	private void emitCountingData(OutputCollector collector) {

//		HBaseStore hStore = new HBaseStore();
		SdsAbnormalLog testBean = new SdsAbnormalLog();

		try{
			Logger.getRootLogger().info("######DEBUG:GoraHbaseBolt initstore before " + hStore);
			hStore =  DataStoreFactory.getDataStore("org.apache.gora.hbase.store.HBaseStore",String.class, SdsAbnormalLog.class,new Configuration());
			Logger.getRootLogger().info("######DEBUG:GoraHbaseBolt initstore after " + hStore);
			
			HBaseRecordAdd hbaseInstance = HBaseRecordAdd.getInstance(zooserver);
			String info = String.format("######DEBUG:GoraHbaseBolt MAP[%d]",fixedSourceMap.size());
			Logger.getRootLogger().info(info);

			for(Entry<String, Integer> entry : fixedSourceMap.entrySet()) {
				
				String entryKey = entry.getKey();
				String[] tuplesp = entryKey.split("\\|",-1);
				String dip = tuplesp[0];
				String sip = tuplesp[1];
				String dport = tuplesp[2];
				String gis = tuplesp[3];
				String idc = tuplesp[4];

				String[] gisArray = gis.split("#");								
				String destAreaName = gisArray[0];
				String destGis = gisArray[1];
//				String sourceAreaName = gisArray[2];
//				String sourceGis = gisArray[3];
//				String sourceAreaCountry = gisArray[4];
//				String sourceAreaId = gisArray[5];		
//				String sourceProvinceName =  gisArray[6].trim(); //如果省为空，精确到国家
//				String sourceAreaCityId = gisArray[7];
//				String sourceAreaProvinceId = gisArray[8];

				String accesstime = new SimpleDateFormat("yyyyMMddHHmmss").format(new Date());
				String rowKeyAbnormal = dip + "_" + accesstime + "_" + sip + "_" + dport + "_9";	
				
				Logger.getRootLogger().info("######DEBUG:GoraHbaseBolt begin to set testBean " + testBean.getFieldsCount());
				testBean.setABRNORMAL("9");
				testBean.setACCESSTIME(new Date().getTime());
				testBean.setATTNUM(upStreampacket.get(entryKey));
				testBean.setDESTAREA(destAreaName);
				testBean.setDESTGEO(destGis);
				testBean.setEVALUATE(60);
				testBean.setSOURCEPORT(Integer.getInteger(dport));
//				Logger.getRootLogger().info("######DEBUG:GoraHbaseBolt SdsAbnormalLogBean is " + testBean.getACCESSTIME());
//				Logger.getRootLogger().info("######DEBUG:GoraHbaseBolt hStore is " + hStore.getClass().getCanonicalName() +  "  " +hStore.toString());
				
				hStore.put(rowKeyAbnormal, testBean);
				Logger.getRootLogger().info("######DEBUG:GoraHbaseBolt after put  ");

			}
			hStore.flush();
		}catch(Exception e){
			e.printStackTrace();
		}finally{
			fixedSourceMap.clear();
		}
	}

	@Override
	public void cleanup() {
		// TODO Auto-generated method stub
		hStore.close();
	}


	@Override
	public void execute(Tuple tuple) {
		// TODO Auto-generated method stub
		try {

			if(isTickTuple(tuple)) {
				emitCountingData(collector);  
				collector.ack(tuple);
			} else {
				countInLocal(tuple); 
				collector.ack(tuple);
			}
		} catch (Exception e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
			collector.fail(tuple);
			throw new FailedException("GoraHbaseBoltBolt throws an exception!");
		}	
	}


	@Override
	public void Init(Map stormConf, TopologyContext context,
			OutputCollector collector) {
//		try {
//			Logger.getRootLogger().info("######DEBUG:GoraHbaseBolt init before " + hStore);
//			hStore =  DataStoreFactory.getDataStore("org.apache.gora.hbase.store.HBaseStore",String.class, SdsAbnormalLog.class,new Configuration());
//			Logger.getRootLogger().info("######DEBUG:GoraHbaseBolt init ...  hstore" + hStore);
//		} catch (Exception e) {
//			// TODO Auto-generated catch block
//			e.printStackTrace();
//		}
	}

}
